package cn.itcast.kafka.optimize;
import cn.itcast.kafka.config.KafkaConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerTestOptimize {

    public static void main(String[] args) throws InterruptedException {
        // 1. 从配置工具类获取配置（无需硬编码）
        Properties props = KafkaConsumerConfig.getConsumerProperties();

        // 2. 创建Kafka消费者实例
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);

        // 3. 订阅要消费的主题（可根据需求修改主题名）
        kafkaConsumer.subscribe(Arrays.asList("test1"));

        // 4. 无限循环拉取消息（消费者需长期运行）
        while (true) {
            // 拉取消息：超时时间5秒（没消息时阻塞，直到超时或有消息）
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));

            // 5. 遍历处理每一条消息
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                String topic = consumerRecord.topic();
                long offset = consumerRecord.offset();
                String key = consumerRecord.key();
                String value = consumerRecord.value();

                // 打印消息详情
                System.out.printf("topic: %s | offset: %d | key: %s | value: %s%n",
                        topic, offset, key, value);
            }

            // 控制消费速率（可选，根据业务需求调整）
            Thread.sleep(1000);
        }
    }
}